package com.rem.kafka.consumer.offset;

import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.LoggerContext;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.Set;

/**
 * 指定offset消费
 *
 * @author Rem
 * @date 2023-01-04
 */
public class CustomConsumerSeek {
    static {
        LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
        Logger root = loggerContext.getLogger("root");
        root.setLevel(Level.INFO);
    }

    public static void main(String[] args) {
        Properties properties = new Properties();
        //连接kafka
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093,localhost:9094,localhost:9095");
        //消费组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group10");

        /**
         * 当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在  （如，数据被删除了），该如何处理？
         * earliest：自动重置偏移量到最早的偏移量。
         * latest：默认，自动重置偏移量为最新的偏移量。
         * none：如果消费组原来的（previous）偏移量不存在，则向消费者抛异常。
         * anything：如果未找到消费者组的先前偏移量，则向消费者抛出异常。
         */
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        //创建消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties, new StringDeserializer(), new StringDeserializer());

        //创建消费者
        kafkaConsumer.subscribe(Collections.singletonList("first"));


        Set<TopicPartition> assignment = kafkaConsumer.assignment();
        while (assignment.size() == 0) {
            kafkaConsumer.poll(Duration.ofSeconds(1));
            /**
             * 获取当前分配给此消费者的分区集。如果订阅是通过使用assign(Collection)直接分配分区发生的，那么这将简单地返回分配的相同分区。
             * 如果使用了主题订阅，那么这将给出当前分配给消费者的主题分区集（如果分配尚未发生，或者分区正在重新分配过程中，则可能没有
             */
            assignment = kafkaConsumer.assignment();
        }
        // 遍历所有分区，并指定 offset 从 50 的位置开始消费
        for (TopicPartition tp : assignment) {
            /**
             * 盖消费者将在下一次poll(timeout)上使用的获取偏移量。如果为同一个分区多次调用此 API，
             * 则最新的偏移量将用于下一个 poll()。请注意，如果在消费过程中任意使用此API，您可能会丢失数据，以重置获取偏移量
             */
            kafkaConsumer.seek(tp, 50);
        }
        //消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
            consumerRecords.forEach(System.out::println);
        }
    }
}
